package net.wicp.tams.commons.binlog.event;

import java.math.BigDecimal;
import java.sql.Timestamp;
import java.util.Date;

import org.apache.commons.lang.ArrayUtils;

import lombok.extern.slf4j.Slf4j;
import net.wicp.tams.commons.apiext.ByteUtil;
import net.wicp.tams.commons.beans.Host;
import net.wicp.tams.commons.binlog.IBinlogRead;
import net.wicp.tams.commons.binlog.IEventRead;
import net.wicp.tams.commons.binlog.MysqlUtil;
import net.wicp.tams.commons.binlog.bean.EventBean;
import net.wicp.tams.commons.binlog.constant.ColumnType;
import net.wicp.tams.commons.callback.ICol;
import net.wicp.tams.commons.callback.impl.column.ColBigDecimal;
import net.wicp.tams.commons.callback.impl.column.ColBytes;
import net.wicp.tams.commons.callback.impl.column.ColDateSql;
import net.wicp.tams.commons.callback.impl.column.ColDateTime;
import net.wicp.tams.commons.callback.impl.column.ColDouble;
import net.wicp.tams.commons.callback.impl.column.ColFloat;
import net.wicp.tams.commons.callback.impl.column.ColInteger;
import net.wicp.tams.commons.callback.impl.column.ColLong;
import net.wicp.tams.commons.callback.impl.column.ColString;
import net.wicp.tams.commons.callback.impl.column.ColTimeSql;
import net.wicp.tams.commons.callback.impl.column.ColTimestamp;
import net.wicp.tams.commons.io.UnsignedLong;

@Slf4j
public abstract class AbsEvent implements IEventRead {
	protected final EventBean event;

	protected byte[] eventbytes;

	protected int curpos = 0;
	
	public static ThreadLocal<Host> hosts= new ThreadLocal<Host>();

	public AbsEvent(EventBean event) {
		this.event = event;
		eventbytes = event.getBody();
	}

	protected byte[] readSubByte(int length) {
		byte[] ret = ArrayUtils.subarray(eventbytes, curpos, curpos + length);
		curpos += length;
		return ret;
	}

	protected byte readByte() {
		return readSubByte(1)[0];
	}

	protected void skip(int length) {
		curpos += length;
	}

	/**
	 * 读剩余字节
	 * 
	 * @return
	 */
	protected byte[] readRest() {
		byte[] ret = ArrayUtils.subarray(eventbytes, curpos, eventbytes.length);
		return ret;
	}

	/***
	 * If the value is < 251, it is stored as a 1-byte integer. If the value is
	 * ≥ 251 and < (216), it is stored as fc + 2-byte integer. If the value is ≥
	 * (216) and < (224), it is stored as fd + 3-byte integer. If the value is ≥
	 * (224) and < (264) it is stored as fe + 8-byte integer.
	 * 
	 * @return
	 */
	public UnsignedLong readUnsignedLong() {
		final int v = ByteUtil.toUnsigned(readByte());
		if (v < 251)
			return UnsignedLong.valueOf(v);
		else if (v == 251)
			return null;
		else if (v == 252)
			return UnsignedLong.valueOf(ByteUtil.readLongL(readSubByte(2)));
		else if (v == 253)
			return UnsignedLong.valueOf(ByteUtil.readLongL(readSubByte(3)));
		else if (v == 254)
			return UnsignedLong.valueOf(ByteUtil.readLongL(readSubByte(8)));
		else
			throw new RuntimeException("assertion failed, should NOT reach here");
	}

	public void rowlog() {
		long temp = event.getBeginBody() + curpos;
		log.info("当前位置:{}", row(temp));
	}

	// 解析列字段
	@SuppressWarnings("rawtypes")
	protected ICol paseCol(ColumnType colType, final int meta) {
		int length = 0;
		if (colType == ColumnType.STRING && meta > 256) {// TODO 没有测试
			final int meta0 = meta >> 8;
			final int meta1 = meta & 0xFF;
			if ((meta0 & 0x30) != 0x30) { // a long CHAR() field: see #37426
				colType = ColumnType.get((byte) (meta0 | 0x30));
				length = meta1 | (((meta0 & 0x30) ^ 0x30) << 4);
			} else {
				ColumnType meta0Type = ColumnType.get((byte) meta0);

				switch (meta0Type) {
				case SET:
				case ENUM:
				case STRING:
					colType = ColumnType.get((byte) meta0);
					length = meta1;
					break;
				default:
					throw new RuntimeException("不支持的类型: " + colType);
				}
			}
		}
		// TODO 处理空值

		ICol retobj = null;
		switch (colType) {
		case TINY:
			int val = ByteUtil.readIntL(readSubByte(1));
			retobj = new ColInteger(val);
			break;
		case SHORT:
			int val2 = ByteUtil.readIntL(readSubByte(2));
			retobj = new ColInteger(val2);
			break;
		case INT24:
			int val3 = ByteUtil.readIntL(readSubByte(3));
			retobj = new ColInteger(val3);
			break;
		case LONG:
			long val4 = ByteUtil.readLongL(readSubByte(4));
			retobj = new ColLong(val4);
			break;
		case LONGLONG:
			long val5 = ByteUtil.readLongL(readSubByte(8));
			retobj = new ColLong(val5);
			break;
		case FLOAT:
			float val6 = Float.intBitsToFloat((int) ByteUtil.readLongL(readSubByte(4)));
			retobj = new ColFloat(val6);
			break;
		case DOUBLE://
			double val7 = Double.longBitsToDouble(ByteUtil.readLongL(readSubByte(8)));
			retobj = new ColDouble(val7);
			break;
		case YEAR:
			int val8 = MysqlUtil.toYear(ByteUtil.readIntL(readSubByte(1)));
			retobj = new ColInteger(val8);
			break;
		case DATE:
			java.sql.Date val9 = MysqlUtil.toDate(ByteUtil.readIntL(readSubByte(3)));
			retobj = new ColDateSql(val9);
			break;
		case TIME:
			java.sql.Time val10 = MysqlUtil.toTime(ByteUtil.readIntL(readSubByte(3)));
			retobj = new ColTimeSql(val10);
			break;
		case DATETIME:
			Date val11 = MysqlUtil.toDatetime(ByteUtil.readLongL(readSubByte(8)));
			retobj = new ColDateTime(val11);
			break;
		case TIMESTAMP:
			Timestamp val12 = MysqlUtil.toTimestamp(ByteUtil.readLongL(readSubByte(4)));
			retobj = new ColTimestamp(val12);
			break;
		case ENUM:
			int val13 = ByteUtil.readIntL(readSubByte(length));
			retobj = new ColInteger(val13);
			break;
		case SET:
			long val14 = ByteUtil.readLongL(readSubByte(length));
			retobj = new ColLong(val14);
			break;
		case BIT:
			final int bitLength = (meta >> 8) * 8 + (meta & 0xFF);
			byte[] val15 = ByteUtil.toBigEndian(readSubByte(bitLength));
			retobj = new ColBytes(val15);
			break;
		case BLOB:
			final int blobLength = ByteUtil.readIntL(readSubByte(meta));
			byte[] val16 = readSubByte(blobLength);
			retobj = new ColBytes(val16);
			break;
		case NEWDECIMAL:
			final int precision = meta & 0xFF;
			final int scale = meta >> 8;
			final int decimalLength = MysqlUtil.getDecimalBinarySize(precision, scale);
			if (precision < scale)
				throw new IllegalArgumentException("invalid precision: " + precision + ", scale: " + scale);
			BigDecimal val17 = MysqlUtil.toDecimal(precision, scale, readSubByte(decimalLength));
			retobj = new ColBigDecimal(val17);
			break;
		case STRING:
			final int stringLength = length < 256 ? ByteUtil.readIntL(readSubByte(1))
					: ByteUtil.readIntL(readSubByte(2));
			String val18 = ByteUtil.readString(readSubByte(stringLength));
			retobj = new ColString(val18);
			break;
		case VARCHAR:
		case VAR_STRING:
			final int varcharLength = meta < 256 ? ByteUtil.readIntL(readSubByte(1))
					: ByteUtil.readIntL(readSubByte(2));
			String val19 = ByteUtil.readString(readSubByte(varcharLength));
			retobj = new ColString(val19);
			break;
		case TIME2:
			final int value1 = ByteUtil.readIntB(readSubByte(3));
			final int nanos1 = ByteUtil.readIntB(readSubByte((meta + 1) / 2));
			java.sql.Time val20 = MysqlUtil.toTime2(value1, nanos1);
			retobj = new ColTimeSql(val20);
			break;
		case DATETIME2:
			final long value2 = ByteUtil.readLongB(readSubByte(5));
			final int nanos2 = ByteUtil.readIntB(readSubByte((meta + 1) / 2));
			Date val21 = MysqlUtil.toDatetime2(value2, nanos2);
			retobj = new ColDateTime(val21);
			break;
		case TIMESTAMP2:
			final long value3 = ByteUtil.readLongB(readSubByte(4));
			final int nanos3 = ByteUtil.readIntB(readSubByte((meta + 1) / 2));
			Timestamp val22 = MysqlUtil.toTimestamp2(value3, nanos3);
			retobj = new ColTimestamp(val22);
			break;
		default:
			log.error("不支持的类型：[{}]", colType.name());
		}
		return retobj;
	}

	///////////////////////////////////////////////////////////////////////////////
	public static String row(Long num) {
		String startstr = String.format(IBinlogRead.BIN_ROW, num, Long.toHexString(num));
		return startstr;
	}

	public static String row(Integer num) {
		String startstr = String.format(IBinlogRead.BIN_ROW, num, Integer.toHexString(num));
		return startstr;
	}

}
